package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.util.LongAccumulator
import org.apache.spark.{SparkConf, SparkContext}

object Demo25Acc {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf()
    conf.setMaster("local")
    conf.setAppName("Demo25Acc")

    val sc: SparkContext = new SparkContext(conf)

    val intRDD: RDD[Int] = sc.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8, 9), 3)

    var l: Long = 100L
    // 累加器在Driver端定义
    val longAcc: LongAccumulator = sc.longAccumulator("longAcc")
    longAcc.add(100)

    intRDD.foreach(i => {
      // 外部变量是在Driver端定义的，算子内部的代码实际上是在Task中执行的
      // 在算子内部用到的外部变量，实际上是一个变量副本，所以操作并不会影响外部变量
      l = l + i
      // 如果需要在算子内部对外部的变量进行简单的累加操作时，可以使用累加器
      // 在算子内部进行累加操作，默认初始值为0
      longAcc.add(i)
      println(l)
      println(s"累加器的值:${longAcc.value}")
    })

    println("Task执行结束")
    println(s"Long类型变量l的值为：$l")
    // 累加器最终会在Driver端进行汇总，会将每个分区对应的Task中的累加器值进行汇总
    println(s"累加器最终的值为：${longAcc.value}")


  }

}
